package com.shujia.flink.tf

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._

object Demo3Filter {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //读取数据
    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")

    //scala api
    val scalaDS: DataStream[String] = studentDS.filter(stu => {
      val clazz: String = stu.split(",")(4)
      "文科一班".equals(clazz)
    })
    //scalaDS.print()


    //java api
    val javaDS: DataStream[String] = studentDS.filter(new FilterFunction[String] {
      override def filter(stu: String): Boolean = {
        val clazz: String = stu.split(",")(4)
        "文科二班".equals(clazz)
      }
    })
    javaDS.print()

    env.execute()

  }

}
